package com.shux.kafka.consumer;

import java.util.concurrent.BlockingQueue;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月20日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 * @param <T>
 */
public class FetchMessageThread<T> extends RepeatThread{
	private BlockingQueue<KafkaMessage<T>> fetchMessageQueue;
	private KafkaConsumer<String, T> consumer;
	public FetchMessageThread(Integer sleepSecords) {
		super(sleepSecords);
	}
	public FetchMessageThread(Integer sleepSecords,BlockingQueue<KafkaMessage<T>> fetchMessageQueue,KafkaConsumer<String, T> consumer) {
		super(sleepSecords);
		this.fetchMessageQueue = fetchMessageQueue;
		this.consumer = consumer;
	}

	@Override
	protected void repeatWork() throws InterruptedException {
		//for(;;){
		logger.debug("准备从kafka中获取数据，并且push到队列fetchMessageQueue中");
			ConsumerRecords<String,T> records = consumer.poll(1000);
			records.forEach(data -> {
				logger.debug("receive message:{}",data.value());
				KafkaMessage<T> message = new KafkaMessage<>(data.value(),data.partition(),data.offset());
				QueueUtils.pushDataToQueue(message,fetchMessageQueue);
			});
		//}
		
		//KafkaMessage<T> message = new KafkaMessage<>(data.value(),data.partition(),data.offset());
		//QueueUtils.pushDataToQueue(message,fetchMessageQueue);
	}

}
